package operator.function;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.functions.Function;
import util.RxJavaHelper;

class OnExceptionResumeNextOperator {

    public static void main(String[] args) {
        onExceptionResumeNext();
    }

    /**
     * 当发生错误时，拦截Exception，并通过自定义Observable
     * 不拦截非Exception
     */
    static void onExceptionResumeNext(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
//                emitter.onError(new Throwable("123"));
                emitter.onError(RxJavaHelper.getException());
            }
        })
                .onExceptionResumeNext(new ObservableSource<Integer>() {
                    @Override
                    public void subscribe(Observer<? super Integer> observer) {
                        observer.onNext(3);
                        observer.onComplete();
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());
    }
}
